package com.csw.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo20Partition {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")

      /**
        * spark shuffle之后rdd默认分区数
        */
      .set("spark.default.partition", "6")
      .setAppName("partition")


    val sc: SparkContext = new SparkContext(conf)

    /**
      * 第一个rdd的分区数默认等于切片数，可以指定一个最小分区数，默认的分区数是2
      */
    val linesRDD: RDD[String] = sc.textFile("spark/data/word.txt", 2)

    println("linesRDD:" + linesRDD.getNumPartitions)

    /**
      * 没有shuffle算子---->后面的rdd分区数默认等于父rdd分区数
      */
    val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(","))

    println("wordsRDD:" + wordsRDD.getNumPartitions)

    val mapRDD: RDD[(String, Int)] = wordsRDD.map((_,1))

    println("mapRDD:"+mapRDD.getNumPartitions)

    /**
      *
      * countRDD 分区数决定因素
      *
      * 1、如果没有指定分区数则默认等于父rdd的分区数
      * 2、shuffle类的算子可以手动指定分区数据（reduce的数量）
      *
      * 分区数决定优先级： 手动指定--->默认分区数（spark.default.partition）--->父rdd分区数
      *
      */
    val countRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_+_,10)

    println("countRDD:"+countRDD.getNumPartitions)

    /**
      * repartition：改变rdd分区数，会产生shuffle，不会做任何逻辑
      *
      * 可以用于通过并行度
      */
    val repartitionRDD: RDD[(String, Int)] = countRDD.repartition(8)

    println("repartitionRDD:"+repartitionRDD.getNumPartitions)

    repartitionRDD.foreach(println)

  }
}
